package edu.utexas.mapreduce;

import java.io.BufferedWriter;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;


public class ReduceOutputWriter implements OutputWriter {

	final String outputDir;
	PrintWriter out;
	
	ReduceOutputWriter(String outputDir, Integer keyID) throws IOException {
		this.outputDir = outputDir;
		String filepath = outputDir + "/" + getFileName(keyID);
		OutputStreamWriter os = new OutputStreamWriter(new FileOutputStream(filepath, false), XML.ENCODING);
		this.out = new PrintWriter(new BufferedWriter(os));
		
		// write XML header and DTD
		out.println("<?xml version=\"1.0\" encoding=\"" + XML.ENCODING + "\"?>");
		out.println("<!DOCTYPE reduce_output [");
		out.println("  <!ELEMENT reduce_output (reduce)*>");
		out.println("  <!ELEMENT reduce (key, value)>");
		out.println("  <!ELEMENT key (#PCDATA)>");
		out.println("  <!ELEMENT value (#PCDATA)>");
		out.println("]>");
		out.println("<reduce_output>");
	}
	
	public void write(String key, String value) {
		out.write("<reduce>");
		out.write("<key><![CDATA[" + key + "]]></key>");
		out.write("<value><![CDATA[" + value + "]]></value>");
		out.write("</reduce>");
		out.println();
	}
	
	void close() {
		out.println("</reduce_output>");
		out.flush();
		out.close();
	}
	
	String getFileName(Integer keyID) {
		return "reduce_" + String.format("%04d", keyID);
	}
}
